/**
 * This file is a part of qloudgen-sec. 
 * You can redistribute qloudgen-sec and/or modify it under the terms of the Lesser GNU General Public License version 3. 
 * qloudgen-sec is distributed WITHOUT ANY WARRANTY. 
 *
 * See the Lesser GNU General Public License for more details at http://www.gnu.org/licenses/. 
 **/
package org.redisson;

import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.HashSet;
import java.util.List;
import java.util.ArrayList;

import org.redisson.cluster.ClusterConnectionManager;
import org.redisson.core.RFakeMap;

import com.lambdaworks.redis.RedisConnection;

public class ClusterRedissonFakeMap<K, V> extends ClusterRedissonFakeExpirable implements RFakeMap<K, V> {

	ClusterRedissonFakeMap(ClusterConnectionManager connectionManager, String name) {
		super( connectionManager , name );
	}

	private List<K> decode(List<K> list , boolean hasCursor) {
		List<K> result = new ArrayList<K>();
		for ( int i = 0 ; i < list.size() ; i++ ) {
			if ( hasCursor && i == 0 ) {
				result.add( list.get( i ) );
			}
			else {
				result.add( decodeKey( list.get( i ) ) );
			}
		}
		return result;
	}
	
	public List<K> scan(final int num , final long cursor , final String pattern , final long count) {
		if ( num < 0 || num > connectionManager.size() - 1 ) {
			return null;
		}
		//
		return decode(
			new ClusterCommand<List<K>>( connectionManager ) {
				@Override
				public List<K> execute(RedisConnection connection) {
					return ( List<K> )connection.scan( cursor , encodeKey( pattern ) , count );
				}
			}.run( num ) , true
		);
	}
	
	@Override
	public Set<K> keySet() {
		Set<K> res = new HashSet<K>();
		for ( int i = 0 ; i < connectionManager.size() ; i++ ) {
			List<K> tmp = new ClusterCommand<List<K>>( connectionManager ) {
				@Override
				public List<K> execute(RedisConnection connection) {
					return ( List<K> )connection.keys( encodeKey( "*" ) );
				}
			}.run( i );
			//
			res.addAll( decode( tmp , false ) );
		}
		return res;
	}

	@Override
	public boolean containsKey(final Object key) {
		return new ClusterCommand<Boolean>( connectionManager ) {
			@Override
			public Boolean execute(RedisConnection connection) {
				return connection.exists( encodeKey( key ) );
			}
		}.run( encodeKey( key ) );
	}

	@Override
	public V get(final Object key) {
		return ( V ) new ClusterCommand<V>( connectionManager ) {
			@Override
			public V execute(RedisConnection connection) {
				return ( V ) connection.get( encodeKey( key ) );
			}
		}.run( encodeKey( key ) );
	}

	@Override
	public V put(final K key, final V value) {
		return ( V ) new ClusterCommand<V>( connectionManager ) {
			@Override
			public V execute(RedisConnection connection) {
				V prev = ( V )connection.get( encodeKey( key ) );
				connection.set( encodeKey( key ) , value );
				return prev;
			}
		}.run( encodeKey( key ) );
	}

	@Override
	public V putIfAbsent(final K key, final V value) {
		return ( V ) new ClusterCommand<V>( connectionManager ) {
			@Override
			public V execute(RedisConnection connection) {
				while ( true ) {
					Boolean res = connection.setnx( encodeKey( key ) , value );
					if ( ! res ) {
						V result = ( V )connection.get( encodeKey( key ) );
						return result;
					}
					else {
						return null;
					}
				}
			}
		}.run( encodeKey( key ) );
	}

	@Override
	public V remove(final Object key) {
		return ( V ) new ClusterCommand<V>( connectionManager ) {
			@Override
			public V execute(RedisConnection connection) {
				V prev = ( V )connection.get( encodeKey( key ) );
				connection.del( encodeKey( key ) );
				return prev;
			}
		}.run( encodeKey( key ) );
	}

    private boolean isEquals(RedisConnection<Object, Object> connection, Object key, Object value) {
        Object val = connection.get(  encodeKey( key ) );
        return ( value != null && value.equals( val ) ) || ( value == null && val == null );
    }

	@Override
	public boolean remove(final Object key, final Object value) {
		return new ClusterCommand<Boolean>( connectionManager ) {
			@Override
			public Boolean execute(RedisConnection connection) {
				while ( true ) {
					connection.watch( encodeKey( key ) );
					if ( connection.exists( encodeKey( key ) ) && isEquals( connection , key , value ) ) {
						connection.multi();
						connection.del( encodeKey( key ) );
						if ( connection.exec().size() == 1 ) {
							return true;
						}
					}
					else {
						connection.unwatch();
						return false;
					}
				}
			}
		}.run( encodeKey( key ) );
	}

	@Override
	public V replace(final K key, final V value) {
		return ( V ) new ClusterCommand<V>( connectionManager ) {
			@Override
			public V execute(RedisConnection connection) {
				while ( true ) {
					connection.watch( encodeKey( key ) );
					if ( connection.exists( encodeKey( key ) ) ) {
						V prev = ( V )connection.get( encodeKey( key ) );
						connection.multi();
						connection.set( encodeKey( key ) , value );
						if ( connection.exec().size() == 1 ) {
							return prev;
						}
					}
					else {
						connection.unwatch();
					}
					return null;
				}
			}
		}.run( encodeKey( key ) );
	}

	@Override
	public boolean replace(final K key, final V oldValue, final V newValue) {
		return new ClusterCommand<Boolean>( connectionManager ) {
			@Override
			public Boolean execute(RedisConnection connection) {
				while ( true ) {
					connection.watch( encodeKey( key ) );
					if ( connection.exists( encodeKey( key ) ) && isEquals( connection , key , oldValue ) ) {
						connection.multi();
						connection.set( encodeKey( key ) , newValue );
						if ( connection.exec().size() == 1 ) {
							return true;
						}
					}
					else {
						connection.unwatch();
						return false;
					}
				}
			}
		}.run( encodeKey( key ) );
	}

	@Override
	public int size() {
		throw new RuntimeException( "FakeMap does not support size()..." );
	}

	@Override
	public boolean isEmpty() {
		throw new RuntimeException( "FakeMap does not support isEmpty()..." );
	}

	@Override
	public boolean containsValue(final Object value) {
		throw new RuntimeException( "FakeMap does not support containsValue()..." );
	}

	@Override
	public void putAll(final Map<? extends K, ? extends V> map) {
		throw new RuntimeException( "FakeMap does not support putAll()..." );
	}

	@Override
	public void clear() {
		throw new RuntimeException( "FakeMap does not support clear()..." );
	}

	@Override
	public Collection<V> values() {
		throw new RuntimeException( "FakeMap does not support values()..." );
	}

	@Override
	public Set<Map.Entry<K, V>> entrySet() {
		throw new RuntimeException( "FakeMap does not support entrySet()..." );
	}
}